package cn.itcast.hello;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * Author itcast
 * Desc 演示使用Java语言开发SparkCore完成WordCount
 */
public class JavaSparkDemo01 {
    public static void main(String[] args) {
        //0.TODO 准备环境
        SparkConf sparkConf = new SparkConf().setAppName("JavaSparkDemo").setMaster("local[*]");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        jsc.setLogLevel("WARN");

        //1.TODO 加载数据
        JavaRDD<String> fileRDD = jsc.textFile("data/input/words.txt");

        //2.TODO 处理数据-WordCount
        //切割
        /*
        @FunctionalInterface
        public interface FlatMapFunction<T, R> extends Serializable {
          Iterator<R> call(T t) throws Exception;
        }
         */
        //注意:java的函数/lambda表达式的语法:
        // (参数列表)->{函数体}
        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
        //每个单词记为1
        JavaPairRDD<String, Integer> wordAndOneRDD = wordsRDD.mapToPair(word -> new Tuple2<>(word, 1));
        //分组聚合
        JavaPairRDD<String, Integer> wordAndCountRDD = wordAndOneRDD.reduceByKey((a, b) -> a + b);

        //3.TODO 输出结果
        List<Tuple2<String, Integer>> result = wordAndCountRDD.collect();
        //result.forEach(t-> System.out.println(t));
        result.forEach(System.out::println);//方法引用/就是方法转为了函数

        //4.TODO 关闭资源
        jsc.stop();
    }
}
